package jv.redis.datatype.queue;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import redis.clients.jedis.Jedis;
/*
 * 普通队列，不阻塞，没有优先级
 */
public class RedisQueueNormal {
	Jedis jedis;
	volatile boolean flag = false;
	ExecutorService es;
	public RedisQueueNormal(){
		initRedis();
		es = Executors.newCachedThreadPool();
	}
	

	public static void main(String[] args) throws InterruptedException, ExecutionException {
		RedisQueueNormal rqn = new RedisQueueNormal();
		List<Future> lf = new ArrayList<Future>();
		lf.add(rqn.producer());
		lf.add(rqn.consumer());
		for(Future f:lf){
			f.get();
		}
		rqn.close();
		System.out.println("program is finished");
	}
	
	public Future producer(){
		System.out.println("submit producer");
		return es.submit(new Runnable() {
				@Override
				public void run() {
					Jedis myEdis = newRedis();
					myEdis.del("redis:queue");
					for(int i=1;i<=10;i++){
						myEdis.lpush("redis:queue", "test"+i);
						System.out.println("producer:test"+i);
					}
					flag = true;
					System.out.println("producer is finished");
				}
			});
	}
	
	public Future consumer() {
		System.out.println("submit consumer");
		return es.submit(new Runnable() {
			@Override
			public void run() {
				Jedis myEdis = newRedis();
				String rlt = "";
				/**
				 * 消费者线程每隔一秒尝试从队列中取一个消息，当取不到消息，且生产者已经完成所有生产，则消费者退出
				 * */
				while (true) {
					rlt = myEdis.rpop("redis:queue");
					if(rlt==null&&flag){
						break;
					}
					System.out.println("consumer:" + rlt);
					try {
						Thread.sleep(1000);
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
				System.out.println("consumer is finished");
			}
		});
	}
	
	public void initRedis(){
		jedis = new Jedis("redis://192.168.1.101:6379");
		System.out.println("redis is connected");
	}
	
	public Jedis newRedis(){
		return new Jedis("redis://192.168.1.101:6379");
	}

	public void closeRedis(){
		//注意不要使用jedis.shutdown()，这样会把服务器停掉。。。
		jedis.close();
	}
	
	public void closePool(){
		es.shutdown();
	}
	
	public void close(){
		closeRedis();
		closePool();
	}
}
